package com.hujf.flink.DataStreamAPI;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;

/**
 * @Author hujf
 * @Date 2022/5/18 16:58
 * @Description TODO
 */
public class AccumulatorDemo {
    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment environment =ExecutionEnvironment.getExecutionEnvironment();

        DataSet<String> input = environment.fromElements("BMW","TSL","Rolls-Ro");

        DataSet<String> result = input.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                intCount.add(1);
                return value;
            }
           IntCounter intCount = new IntCounter();
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);

                getRuntimeContext().addAccumulator("myCount",intCount);
            }
        });

        result.print();
        JobExecutionResult execute = environment.execute();

        int mycount = execute.getAccumulatorResult("myCount");

        System.out.println(mycount);
    }
}
